package com.hujinwen.utils.server;

import org.apache.http.HttpHost;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.core.CountRequest;
import org.elasticsearch.client.core.CountResponse;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.builder.SearchSourceBuilder;

import java.io.Closeable;
import java.io.IOException;
import java.util.List;

/**
 * Created by joe on 2020/3/13
 * <p>
 * es操作的封装
 */
public class ESClient implements Closeable {

    private RestHighLevelClient client;

    public ESClient(HttpHost... hosts) {
        if (hosts.length == 0) {
            hosts = new HttpHost[]{new HttpHost("localhost", 9200)};
        }
        client = new RestHighLevelClient(RestClient.builder(hosts));
    }


    /**
     * 创建index
     */
    public boolean createIndex(String index) throws IOException {
        final CreateIndexRequest request = new CreateIndexRequest(index);
        final CreateIndexResponse response = client.indices().create(request, RequestOptions.DEFAULT);
        return response.isAcknowledged();
    }

    /**
     * 查询index是否存在
     */
    public boolean exist(String... indices) throws IOException {
        return client.indices().exists(new GetIndexRequest(indices), RequestOptions.DEFAULT);
    }

    /**
     * 查询index记录总数
     */
    public long getTotal(String... indices) throws IOException {
        final CountRequest countRequest = new CountRequest(indices);

        final CountResponse response = client.count(countRequest, RequestOptions.DEFAULT);

        return response.getCount();
    }

    /**
     * 搜索全部
     */
    public String searchAll(String... indices) throws IOException {
        final SearchRequest searchRequest = new SearchRequest(indices);

        final SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(QueryBuilders.matchAllQuery());

        searchRequest.source(searchSourceBuilder);

        final SearchResponse resp = client.search(searchRequest, RequestOptions.DEFAULT);
        return resp.toString();
    }


    /**
     * match query
     */
    public String matchQuery(String name, String value, String... indices) throws IOException {
        final SearchRequest searchRequest = new SearchRequest(indices);

        final SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(QueryBuilders.matchQuery(name, value));

        searchRequest.source(searchSourceBuilder);
        final SearchResponse resp = client.search(searchRequest, RequestOptions.DEFAULT);
        return resp.toString();
    }


    /**
     *
     */
    public String boolQuery(String... indices) throws IOException {
        final BoolQueryBuilder boolQuery = QueryBuilders.boolQuery()
                .must(QueryBuilders.matchQuery("anwser", "摇滚歌手"))
                .must(QueryBuilders.matchQuery("question", "中国"));

        final SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(boolQuery);

        final SearchRequest searchRequest = new SearchRequest(indices);
//        searchRequest.searchType("");

        final SearchResponse resp = client.search(searchRequest, RequestOptions.DEFAULT);

        return resp.toString();
    }

    public void postData(String indexName, String jsonStr) {
        try {
            // 1、创建索引请求  //索引  // mapping type  //文档id
            IndexRequest request = new IndexRequest(indexName);     //文档id
            // 2、准备文档数据
            // 直接给JSON串
            request.source(jsonStr, XContentType.JSON);
            //4、发送请求
            IndexResponse indexResponse = null;
            try {
                // 同步方式
                indexResponse = client.index(request, RequestOptions.DEFAULT);
            } catch (ElasticsearchException e) {
                // 捕获，并处理异常
                //判断是否版本冲突、create但文档已存在冲突
                if (e.status() == RestStatus.CONFLICT) {
                    System.out.println("冲突了，请在此写冲突处理逻辑！" + e.getDetailedMessage());
                }
            }
            //5、处理响应
            if (indexResponse != null) {
                String index1 = indexResponse.getIndex();
                String type1 = indexResponse.getType();
                String id1 = indexResponse.getId();
                long version1 = indexResponse.getVersion();
                if (indexResponse.getResult() == DocWriteResponse.Result.CREATED) {
                    System.out.println("新增文档成功!" + index1 + type1 + id1 + version1);
                } else if (indexResponse.getResult() == DocWriteResponse.Result.UPDATED) {
                    System.out.println("修改文档成功!");
                }
                // 分片处理信息
                ReplicationResponse.ShardInfo shardInfo = indexResponse.getShardInfo();
                if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
                    System.out.println("分片处理信息.....");
                }
                // 如果有分片副本失败，可以获得失败原因信息
                if (shardInfo.getFailed() > 0) {
                    for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                        String reason = failure.reason();
                        System.out.println("副本失败原因：" + reason);
                    }
                }
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 批量插入（必要的时候设成异步的，能提高效率?）
     */
    public void batchPost(String indexName, List<String> jsonList) throws IOException {
        final BulkRequest bulkRequest = new BulkRequest(indexName);

        for (String jsonStr : jsonList) {
            final IndexRequest indexRequest = new IndexRequest();
            indexRequest.source(jsonStr, XContentType.JSON);
            bulkRequest.add(indexRequest);
        }

        client.bulk(bulkRequest, RequestOptions.DEFAULT);
//        client.bulkAsync(bulkRequest, RequestOptions.DEFAULT, null);
    }


    public void close() throws IOException {
        client.close();
    }
}
